package com.asap.demo.function;

import com.asap.demo.model.BeanField;
import com.asap.demo.rete.ReteDemo4;
import com.asap.demo.table.JsonField;
import com.asap.demo.table.RuleParse;
import com.asap.demo.utils.DateUtil;
import com.asap.demo.utils.Utils;
import com.asap.rule.StandardEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Field;
import java.time.Instant;
import java.util.Date;

/**
 * @author wangbh
 * @Description: TODO
 * @date 2021/11/15 10:21
 */
public class dealMapFunction implements MapFunction<StandardEvent, BeanField> {
    private static final Logger logger = LoggerFactory.getLogger(dealMapFunction.class);
    RuleParse ruleParse;

    public dealMapFunction(RuleParse ruleParse) {
        this.ruleParse = ruleParse;
    }

    @Override
    public BeanField map(StandardEvent standardEvent) throws Exception {
        //构造对象属性
        BeanField beanField = new BeanField();
        // 获取obj类的字节文件对象
        Class classInfo = beanField.getClass();

        logger.info("size:" + ruleParse.getTableDefine().getJsonFieldList().size());
        logger.info("ruleId"+standardEvent.getField("rule_id"));
        beanField.setRuleId(standardEvent.getField("rule_id"));
        //给实例对象赋值
        for (Object jsonField : ruleParse.getTableDefine().getJsonFieldList()) {
            JsonField jsonFieldBean = (JsonField) jsonField;
            String field = Utils.getUrlByCode(jsonFieldBean.getDbFieldName().toLowerCase());
            if(field.equals("alertsignatureIdL")){
                field="alertSignatureIdL";
            }
            if (jsonFieldBean.getFieldType().equals("date")) {
                String fieldValue = standardEvent.getField(jsonFieldBean.getTopicFieldName());
                if (fieldValue != null) {
                    Date date = Utils.transforDateForDate(fieldValue);
                    if (date != null) {
                        long longDate = date.getTime();
                        Utils.setFieldValueByName(classInfo,beanField, field, Instant.ofEpochMilli(longDate));
                    }
                }

            } else if (jsonFieldBean.getFieldType().equals("int")) {
                String fieldValueStr = standardEvent.getField(jsonFieldBean.getTopicFieldName());
                Integer fieldValue = 0;
                if (!(fieldValueStr == null || fieldValueStr.isEmpty())) {
                    if (!Utils.checkAllNumber(fieldValueStr)) {
                        fieldValue = Double.valueOf(fieldValueStr).intValue();
                    }
                }
                Utils.setFieldValueByName(classInfo,beanField, field, fieldValue);
            } else if (jsonFieldBean.getFieldType().equals("long")) {
                String fieldValueStr = standardEvent.getField(jsonFieldBean.getTopicFieldName());
                Long fieldValue = 0L;
                if (!(fieldValueStr == null || fieldValueStr.isEmpty())) {
                    if (!Utils.checkAllNumber(fieldValueStr)) {
                        fieldValue = Double.valueOf(fieldValueStr).longValue();
                    }
                }
                Utils.setFieldValueByName(classInfo,beanField, jsonFieldBean.getDbFieldName(), fieldValue);

            }else if (jsonFieldBean.getFieldType().equals("string")) {
                String fieldValueStr = new String("");
                fieldValueStr = standardEvent.getField(jsonFieldBean.getTopicFieldName());
                if(fieldValueStr == null){
                    fieldValueStr ="";
                }
                Utils.setFieldValueByName(classInfo,beanField, field, fieldValueStr);
            }
        }
        return beanField;
    }
}
